package cn.com.netedge.iot.common.utils;

import cn.com.netedge.iot.common.constants.MqConstants;
import com.alibaba.fastjson2.JSON;
import net.openhft.chronicle.bytes.MethodReader;
import net.openhft.chronicle.queue.ExcerptAppender;
import net.openhft.chronicle.queue.ExcerptTailer;
import net.openhft.chronicle.queue.RollCycles;
import net.openhft.chronicle.queue.impl.single.SingleChronicleQueue;
import net.openhft.chronicle.queue.impl.single.SingleChronicleQueueBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ChronicleQueueUtil {
    private static final Logger logger = LoggerFactory.getLogger(ChronicleQueueUtil.class);
    public static final Map<String, SingleChronicleQueue> queueMap = new ConcurrentHashMap<>();
    private static final Map<String, Lock> lockMap = new ConcurrentHashMap<>();

    public interface MessageConsumer {
        void onMessage(String json);
    }

    // Method to get or create a queue
    private static SingleChronicleQueue getQueue(String queueName) {
        return queueMap.computeIfAbsent(queueName, name -> {
            SingleChronicleQueue queue = SingleChronicleQueueBuilder
                    .binary(new File(name + ".cq"))
                    .rollCycle(RollCycles.FAST_DAILY)
                    .build();
            lockMap.putIfAbsent(queueName, new ReentrantLock());
            MqConstants.putMessageCount(queueName,  new AtomicLong(0));
            return queue;
        });
    }

    public static long enqueue(String queueName, Object message) {
        SingleChronicleQueue queue = getQueue(queueName);
        Lock lock = lockMap.get(queueName);
        if (lock != null) {
            lock.lock();  // 加锁
            // Get the queue
            // Get an appender for the queue
            try (ExcerptAppender appender = queue.acquireAppender()) {
                // Get a method writer for the MessageConsumer interface
                MessageConsumer writer = appender.methodWriter(MessageConsumer.class);
                // Call the onMessage method to write the JSON string to the queue
                if (message instanceof String) {
                    writer.onMessage((String) message);
                } else {
                    writer.onMessage(JSON.toJSONString(message));
                }
                return increment(queueName);
            } catch (Exception e) {
                logger.info("send ChronicleQueue [{}] message error.", queueName, e);
            } finally {
                lock.unlock();  // 释放锁
            }
        }
        return -1;
    }

    public static String dequeue(String queueName) {
        SingleChronicleQueue queue = getQueue(queueName);
        Lock lock = lockMap.get(queueName);
        if (lock != null) {
            lock.lock();  // 加锁
            try {
                // Get the queue
                if (queue.isClosed()) {
                    return null;
                }

                // Get a tailer for the queue
                try (ExcerptTailer tailer = queue.createTailer("tailer")) {
                    MessageHolder holder = new MessageHolder();
                    MethodReader reader = tailer.methodReader(holder);

                    if (queue.isClosed()) {
                        return null;
                    }
                    if (reader.readOne()) {
                        String result = holder.getMessage();
                        decrement(queueName);
                        return result;
                    }

                } // tailer is closed
                return null;
            } catch (Exception e) {
                logger.info("dequeue error", e);
            } finally {
                lock.unlock();  // 释放锁
            }
        }
        return null;
    }

    // Holder class for the message
    private static class MessageHolder implements MessageConsumer {
        private String message;

        @Override
        public void onMessage(String message) {
            this.message = message;
        }

        public String getMessage() {
            return message;
        }
    }

    // Make sure to close all queues
    public static void close() {
        Iterator<Map.Entry<String, SingleChronicleQueue>> entryIt = queueMap.entrySet().iterator();
        Map.Entry<String, SingleChronicleQueue> entry;
        String queueName;
        SingleChronicleQueue queue;
        while (entryIt.hasNext()) {
            entry = entryIt.next();
            queue = entry.getValue();
            queueName = entry.getKey();
            Lock lock = lockMap.get(queueName);
            if (lock != null) {
                lock.lock();  // 加锁
                try {
                    queue.close();
                } catch (Exception e) {
                    logger.info("close chronicle queue error.", e);
                } finally {
                    lock.unlock();  // 释放锁
                }
            }
        }
    }

    // 消息生产时 +1 操作
    public static long increment(String queueName) {
        // 如果 queueName 不存在，初始化为 0，否则对其计数进行 +1
        AtomicLong count = MqConstants.getMessageCount(queueName);
        long messageCount = count.incrementAndGet();
        logger.debug("chronicle queue message count, after enqueue: {} -> {} ", queueName, messageCount);
        return messageCount;
    }

    // 消息消费时 -1 操作
    public static long decrement(String queueName) {
        AtomicLong count = MqConstants.getMessageCount(queueName);
        long messageCount = count.decrementAndGet();
        logger.debug("chronicle queue message count, after dequeue: {} -> {} ", queueName, messageCount);
        return messageCount;
    }


    public static void main(String[] args) {
        String queueName = "test04";
        // Enqueue a message
        for (int i = 0; i < 10; i++) {
            String value = "Hello World" + i;
            long messageCount = enqueue(queueName, value);
            System.out.println("enqueue count: " + messageCount);
        }

        // Dequeue a message

        String message = dequeue(queueName);
        System.out.println("dequeue count: " + MqConstants.getMessageCount(queueName));
        int i = 0;
        while (message != null) {
            System.out.println("dequeue count: " + MqConstants.getMessageCount(queueName));
            i++;
            System.out.println(message);
            message = dequeue(queueName);
        }
    }
}
